package com.aotain.nyx.synflood;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

public class SynfloodFixedStatMap implements MapFunction<Tuple2<String,String>, Tuple2<String,SynfloodTuple>> {

	/** 
	* @Fields serialVersionUID : TODO(用一句话描述这个变量表示什么) 
	*/ 
	private static final long serialVersionUID = 9021240131597875760L;

	@Override
	public Tuple2<String, SynfloodTuple> map(Tuple2<String, String> value)
			throws Exception {

		Tuple2<String,SynfloodTuple> ret = new Tuple2<String,SynfloodTuple>();
		String logString = value.f1; //格式化后的原始数据
		String[] items = logString.split("\\|",-1);
		if(items.length <= 10)
			return ret;
		String sip = value.f0;
		String dip = items[0];
		String dport = items[1];
		String gis = new String(Base64.decodeBase64(items[4]));;
		String idc = items[9];
		String status = items[12];
		String type = items[7];
		String flowdir = items[19];
		long upStreamOctets = 0; //UPSTREAMOCTETS
		long upStreamPacket = 0; //UPSTREAMPACKET
		String[] gisArray = gis.split("#");								
		String destAreaName = gisArray[0];
		String destGis = gisArray[1];
		String fixedMapKey = dip + "|" + sip + "|" + dport + "|" + gis + "|" + idc + "|isFixed|" +flowdir;
//		String forgedMapKey = dip + "|" + dport + "|" + idc + "|" + destAreaName + "|" + destGis;
		
		upStreamOctets = Long.parseLong(items[10]);
		upStreamPacket = Long.parseLong(items[11]);
		
		if(status == "0" && "6".equals(type)) {
			SynfloodTuple info = new SynfloodTuple();
			info.setUpStreamOctets(upStreamOctets);
			info.setUpStreamPacket(upStreamPacket);
			String retKey = String.format("%s", fixedMapKey);
			
			ret.setFields(retKey, info);
			
		}
		
		return ret;
	}

}
